package com.luojing.springcloud.rocketmq;

import java.io.UnsupportedEncodingException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.stereotype.Component;

/**
 * rocketMq简单的发送消息
 * @author luojing
 * @create 2018/10/23 2:48 PM
 **/
@Component
public class SimpleProducer {

  /**
   * 同步发送消息
   * @throws Exception
   */
  public static void sendSyncMessage() throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer("group_rocket");
    //建立对象,定义组
    try {
      //设定mq地址
      producer.setNamesrvAddr("localhost:9876");
      producer.start();

      for(int i = 0; i < 100; i ++){
        //topic 一级标题,tag 二级标题,同一大业务,内有很多不同的消息类型,使用tag区分,实现精确消费
        Message message = new Message("luojing"/*topic*/,"TagA" /*tag*/,
            ("Hello RocketMq" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /*消息内容*/);
        message.putUserProperty("a","2");
        SendResult sendResult = producer.send(message);
        System.out.printf("%s%n",sendResult);
      }
    } finally {
      producer.shutdown();
    }
  }

  public static void sendAsyncMessage()throws Exception{
    DefaultMQProducer producer = new DefaultMQProducer("group_rocket");

    producer.setNamesrvAddr("localhost:9876");
    producer.start();

    producer.setRetryTimesWhenSendAsyncFailed(0);
    for (int i = 0; i < 100; i++) {
      final int index = i;
      //Create a message instance, specifying topic, tag and message body.
      Message msg = new Message("luojing",
          "TagB",
          "OrderID188",
          "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
      producer.send(msg, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          System.out.printf("%-10d OK %s %n", index,
              sendResult.getMsgId());
        }
        @Override
        public void onException(Throwable e) {
          System.out.printf("%-10d Exception %s %n", index, e);
          e.printStackTrace();
        }
      });
    }
    //这里异步需要让线程休眠,不然直接退出就发不了了
    Thread.sleep(10000L);
    //Shut down once the producer instance is not longer in use.
    producer.shutdown();
  }

  public static void sendOnewayMessage()throws Exception{
    //Instantiate with a producer group name.
    DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    // Specify name server addresses.
    producer.setNamesrvAddr("localhost:9876");
    //Launch the instance.
    producer.start();
    for (int i = 0; i < 1; i++) {
      //Create a message instance, specifying topic, tag and message body.
      Message msg = new Message("TopicTest" /* Topic */,
          "TagA" /* Tag */,
          ("Hello RocketMQ " +
              i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
      );
      //Call send message to deliver message to one of brokers.
      producer.sendOneway(msg);

    }

    //Shut down once the producer instance is not longer in use.
    producer.shutdown();
  }

  public static void main(String[] args) throws Exception {
    //sendSyncMessage();
    //sendAsyncMessage();
    //[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 48]
    sendOnewayMessage();
  }
}
